package ru.ifmo.croak.impl.sharding.team3;

import ru.ifmo.croak.api.*;
import ru.ifmo.croak.util.CroakFactory;

import java.io.Serializable;
import java.util.concurrent.*;

public class ShardingImpl implements PostService, QueryService
{
    final PostService[] postServices;
    final QueryService[] queryServices;
    final ExecutorService executor;

    public ShardingImpl(String componentName) {
        CroakFactory factory = CroakFactory.getInstance();
        String providersString = factory.getProperty(componentName + ".providers");
        String[] providers = providersString.split(",");
        executor = new ThreadPoolExecutor(2, 10, 5,  TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(100),
            new ThreadPoolExecutor.CallerRunsPolicy());

        postServices = new PostService[providers.length];
        queryServices = new QueryService[providers.length];
        for(int i = 0; i < providers.length; ++i) {
            String providerName = providers[i];

            postServices[i] = factory.getImplementation(PostService.class, providerName);
            queryServices[i] = factory.getImplementation(QueryService.class, providerName);
        }
    }

    public ExecutorService getExecutor() {
        return executor;
    }

    int getMessageKey(Message message) {
        return message.getId().hashCode();
    }

    int getShardNo(Message message) {
        int messageKey = getMessageKey(message);
        long intRange = (long) Integer.MAX_VALUE - Integer.MIN_VALUE + 1;
        long perShard = (intRange + postServices.length - 1) / (postServices.length); // ceil
        long keyVal = (long) messageKey - Integer.MIN_VALUE;
        return (int) (keyVal / perShard);
    }


    @Override
    public void postMessage(Message message) {
        int shardNo = getShardNo(message);
        assert shardNo < postServices.length;
        postServices[shardNo].postMessage(message);
    }

    @Override
    public MessageQuery createQuery(QueryRequest request) {
        MessageQuery[] queries = new MessageQuery[queryServices.length];
        for(int i = 0; i < queryServices.length; ++i) {
            queries[i] = queryServices[i].createQuery(request);
            assert queries[i] != null;
        }

        return new CompositeMessageQuery(this, request, queries);
    }
}
